package org.qloudgen.borg.cluster.operator;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.qloudgen.borg.CommonValues;
import org.qloudgen.borg.beans.Cluster;
import org.qloudgen.borg.metadb.DefaultImpl;

public class HadoopClusterOperatorImpl extends AbstractClusterOperator{
	
	static Log log = LogFactory.getLog(HadoopClusterOperatorImpl.class);
	
	public HadoopClusterOperatorImpl(Cluster cluster, String operatorType) {
		super(cluster, operatorType);
	}

	public int start() {
		return clusterAction(cluster,CommonValues.STATUS_ON);
	}

	public int stop(){
		return clusterAction(cluster,CommonValues.STATUS_OFF);	
	}

	@Override
	public int uploadJar(){
		
		try {
			this.getExec().scpFile(this.getCluster().getLocalJar(), this.getCluster().getJarPath());
		} catch (IOException e) {
			log.error(e.getMessage());
			return -1;
		}
		return 0;
	}

	@Override
	public int add() {

		return 0;
	}

	@Override
	public int remove() {

		return 0;
	}
	
	private int clusterAction(Cluster cluster, String status){
		
		String cmd ;
		int version  = Integer.parseInt(cluster.getVersion());
		
		switch (version) {
		
			case 1 : {
					cmd = status.equalsIgnoreCase(CommonValues.STATUS_ON)?"start-all.sh":"stop-all.sh";
					this.getExec().exec(cmd);
					break;
				}
			case 2 : {
					cmd = status.equalsIgnoreCase(CommonValues.STATUS_ON)?"start-dfs.sh":"stop-yarn.sh";
					this.getExec().exec(cmd);
					
					cmd = status.equalsIgnoreCase(CommonValues.STATUS_ON)?"start-yarn.sh":"stop-dfs.sh";
					this.getExec().exec(cmd);
					
					break;
				}		
		}
				
//		if ( StringUtils.isNotEmpty(result) ) return -1;
		
		cluster.setClusterStatus(status);
		DefaultImpl.updateClusterStatus(cluster);
		return 0;
	}
}
